package com.unionpay.async;

import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;

import org.apache.log4j.Logger;

/**
 * 异步任务工作池
 * 
 * @author liuzy
 * @since 2015年6月12日
 */
public class AsyncTaskPool {
	private static Logger logger = Logger.getLogger(AsyncTaskPool.class);

	/**
	 * 线程是否运行
	 */
	private static volatile boolean isRun = true;

	/**
	 * 工作队列大小
	 */
	private static int queueSize = 5000;

	/**
	 * 工作线程的数量
	 */
	private static int runQuantity = 10;

	/**
	 * 队列,此队列按照 FIFO（先进先出）原则对元素进行排序
	 */
	private static Queue<AsyncTask> queue = new ConcurrentLinkedQueue<AsyncTask>();

	/**
	 * 向队列中增加AsyncTask对象
	 */
	public static void add(AsyncTask task) {
		if (null != task && isRun) { // 此队列不允许使用 null 元素
			synchronized (queue) {
				if (queue.size() <= queueSize) {
					queue.offer(task);
				} else {
					queue.poll(); // 获取并移除此队列的头，如果此队列为空，则返回 null
					queue.offer(task); // 将指定元素插入此队列的尾部
				}
			}
		}
	}

	/**
	 * 获取AsyncTask对象
	 */
	private static AsyncTask getAsyncLog() {
		synchronized (queue) {
			if (queue.isEmpty()) {
				return null;
			} else {
				return queue.poll(); // 获取并移除此队列的头，如果此队列为空，则返回 null
			}
		}
	}

	/**
	 * 启动方法
	 */
	public static void start() {
		logger.info("AsyncTaskPool start...");
		try {
			for (int i = 0; i < runQuantity; i++) {
				Thread asyncWorkThread = new Thread(new Runnable() {
					public void run() {
						while (isRun) {
							try {
								AsyncTask task = getAsyncLog();
								if (null == task) {
									Thread.sleep(200);
								} else {
									task.execute();
								}
							} catch (Exception e) {
								e.printStackTrace();
							}
						}
					}
				});
				asyncWorkThread.setName("AsyncLogPool-AsyncWorkThread-" + (i + 1));
				asyncWorkThread.start();
			}
			logger.info("AsyncWorkThread " + runQuantity + " started.");
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	/**
	 * 销毁方法
	 */
	public static void destroy() {
		logger.info("AsyncTaskPool destroy...");
		queue.clear();
		isRun = false;
		try {
			Thread.sleep(500);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
}
